//package cn.ipanel.bigdata.job.basic
//
//import cn.ipanel.bigdata.boot.Job
//import cn.ipanel.bigdata.boot.config.Configuration
//import org.apache.spark.sql.functions.{col, length, lit}
//import org.apache.spark.sql.types.LongType
//import org.apache.spark.sql.{DataFrame, Row, SparkSession}
//import cn.ipanel.bigdata.dw.dim.mysql.T_DEVICE_INFO
//import cn.ipanel.bigdata.dw.dim.mysql.T_DEVICE_INFO._
//
///**
// * @author lzz
// * @environment IntelliJ IDEA 2020.3.1
// * @projectName bigdata_shanxi_git
// * @date 2023/06/16 14:56
// * @description:
// */
//class RedisTest2 extends Job {
//
//  override def optCostumerConf: Map[String, String] = {
//
//
//    val mp = Map("spark.redis.host" -> Configuration.getDatabase("redis_online").host
//      , "spark.redis.port" -> Configuration.getDatabase("redis_online").port
//      , "spark.redis.auth" -> Configuration.getDatabase("redis_online").password
//      , "spark.redis.db" -> Configuration.getDatabase("redis_online").db)
//
//    println("lzz try set: " + mp.toString())
//    mp
//  }
//
//  override def onStartup(): Unit = {
//
//    // 必须保证先本项目的spark来确保先初始化redis的配置到spark，再调用设备任务的spark
//    // 因为本项目的spark和设备任务的spark并不是同一个(但奇怪的是设备任务能读取到本spark的redis配置，把T_DEVICE_INFO.load先执行却读不到)
//    // println(spark)               => org.apache.spark.sql.SparkSession@232cce0
//    // println(T_DEVICE_INFO.spark) => org.apache.spark.sql.SparkSession@405223e4
//    val df1 = loadFromRedis3(spark)
//    df1.show(false)
//
//    val df2 = T_DEVICE_INFO.load
//      .select(F_STB_ID, F_MAC, F_CITY_ID, F_COUNTY_ID, F_TOWN_ID, F_ROAD_ID, F_ZONE_ID)
//      .filter(length(col(F_STB_ID)) > 10)
//    df2.show(false)
////    println(T_DEVICE_INFO.spark.conf.get("spark.redis.host"))
////    println(T_DEVICE_INFO.spark.conf.get("spark.redis.port"))
////    println(T_DEVICE_INFO.spark.conf.get("spark.redis.auth"))
////    println(T_DEVICE_INFO.spark.conf.get("spark.redis.db"))
//
//    df2.join(df1, Seq(F_STB_ID))
//      .show(false)
//  }
//
//
//  def saveToReids(spark: SparkSession): Unit = {
//    val rdd = spark.sparkContext.parallelize(Seq(
//      ("key1", "100"),
//      ("key2", "101"),
//      ("key3", "50")
//    ))
//
//    // 这种存储方式，将产生n多不认识的key
//    // 1) "foo:c16d89b53a194ddd831e45a60e0d574d"
//    // 2) "_spark:foo:schema"
//    // 3) "foo:ba8626441fd5474f9164a9585cf5b983"
//    // 4) "foo:64f61b647799494ebcd35e7c63ca074f"
//    //    df.write
//    //      .format("org.apache.spark.sql.redis")
//    //      .option("table", "foo")
//    //      .save()
//
//    import com.redislabs.provider.redis._
//    // 效果还是可以的
//    // 127.0.0.1:6379[1]> zrange test2 0 100  withscores
//    // 1) "key3"
//    // 2) "50"
//    // 3) "key1"
//    // 4) "100"
//    // 5) "key2"
//    // 6) "101"
//    spark.sparkContext.toRedisZSET(rdd, "test2")
//
//  }
//
//  def loadFromRedis3(spark: SparkSession): DataFrame = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    val rdd = sc.fromRedisZSetWithScore("test", 3)
//    rdd.toDF(F_STB_ID, "timestamp")
//      .union(
//        spark.sparkContext.parallelize(Seq(
//          ("003501FF0001181001FC08A5C8CADD4F", "123"),
//          ("003501FF0001181001FC08A5C8D14093", "123")
//        )).toDF(F_STB_ID, "timestamp")
//      )
//  }
//
//  def loadFromRedis2(spark: SparkSession): Unit = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    val rdd = sc.fromRedisZSetWithScore("test", 3)
//    val df = rdd.toDF("mac", "timestamp")
//    df.printSchema()
//    df.show(false)
//
//    val df2 = df.withColumn("timestamp", col("timestamp").cast(LongType))
//    df2.show()
//
//    println(df2.rdd.getNumPartitions)
//    val df3 = df2.rdd.mapPartitionsWithIndex((partId, iter) => {
//      // 这里foreach打印之后，就不能在后面再show了，会无数据
//      //      iter.foreach(x => {
//      //        println("partitionId: " + partId + " value: " + Tuple2())
//      //      })
//      iter
//        .map(x => Value(x.getAs[String]("mac"), x.getAs[Long]("timestamp"), partId))
//        .toList
//        .sortBy(x => x.timestamp)
//        .iterator
//    })
//      .toDF()
//    df3.show()
//
//    println("partition 0")
//    println(df3.filter(col("partitionId") === lit(0)).count())
//    println("partition 1")
//    println(df3.filter(col("partitionId") === lit(1)).count())
//    println("partition 2")
//    println(df3.filter(col("partitionId") === lit(2)).count())
//  }
//
//
//  def loadFromRedis(spark: SparkSession): Unit = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    // 这种方式只会得到值，而拿不到分数
//    //    val rdd = sc.fromRedisZRange("south:stb:online.info", 0, 1696089600, 3)
//    val rdd = sc.fromRedisZSetWithScore("south:stb:online.info", 3)
//    val df = rdd.toDF("mac", "timestamp")
//    df.printSchema()
//    df.show(false)
//
//    val df2 = df.withColumn("timestamp", col("timestamp").cast(LongType)).repartition(2)
//    df2.show()
//
//
//    println(df2.rdd.getNumPartitions)
//    df2.rdd.mapPartitionsWithIndex((partId, iter) => {
//      // 这里foreach打印之后，就不能在后面再show了，会无数据
////      iter.foreach(x => {
////        println("partitionId: " + partId + " value: " + Tuple2())
////      })
//      iter
//        .map(x => Value(x.getAs[String]("mac"), x.getAs[Long]("timestamp"), partId))
//        .toList
//        .sortBy(x => x.timestamp)
//        .iterator
//    })
//      .toDF()
//      .show()
//  }
//
//}
//
//case class Value(var mac: String,
//                 var timestamp: Long,
//                 var partitionId: Int)
//
//object Value {
//  def apply(t: Row, partId: Int): Value = {
//    Value(
//      t.getAs[String]("mac"),
//      t.getAs[Long]("timestamp"),
//      partId
//    )
//  }
//}